package com.ndp.fb.mq.producer;

import com.alibaba.rocketmq.common.message.Message;
import com.ndp.fb.cache.redis.serializer.KryoRedisSerializer;
import com.ndp.fb.config.ConfigCentre;
import com.ndp.fb.constant.ConfigConst;
import com.ndp.fb.exception.BusinessException;
import com.ndp.fb.exception.ErrorCode;
import com.ndp.fb.mq.api.producer.RocketMqProducerService;
import com.ndp.fb.mq.producer.helper.RocketMqProducerHelper;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * Created by infi on 2015/1/21.
 */
@Component
public class RocketMqProducerServiceImpl implements RocketMqProducerService {
    private final KryoRedisSerializer byteSerializer = new KryoRedisSerializer();
    private Logger logger = LoggerFactory.getLogger(RocketMqProducerServiceImpl.class);
    @Autowired
    private RocketMqProducerHelper rocketMqProducerHelper;

    private String TOP_NAME_PREFIX = ConfigCentre.getString(ConfigConst.MQ_SERVICES_TOPIC_PREFIX);

    @Override
    public void sendMessage(String topicName, String tags, String keys, Object message) {
        topicName = setTopicName(topicName, "准备消息发送 topicName >>> " + topicName + ", tags >>>> " + tags + " , keys>>>>" + keys, keys);
        try {
            Message msg = new Message(topicName, tags, keys,byteSerializer.serialize(message));
            rocketMqProducerHelper.send(msg);
            logger.info("消息发送成功 keys >>>>  " + msg.getKeys());
        } catch (Exception e) {
            logger.info("准备消息发送失败 keys >>>>  " + keys, e);
        }
    }


    public void sendMessage(String topicName, String tags, String keys, Object message, int delayLevel) {
        topicName = setTopicName(topicName, tags, keys);
        try {
            Message msg = new Message(topicName, tags, keys, byteSerializer.serialize(message));
            if (delayLevel < 0 || delayLevel > 18) {
                logger.error("延时等级不在支持范围内。0：不延时；1 - 18：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h。");
                throw new BusinessException(ErrorCode.ERROR_1000);
            } else {
                msg.setDelayTimeLevel(delayLevel);
            }
            rocketMqProducerHelper.send(msg);
            logger.info("消息发送成功 keys >>>>  "+ msg.getKeys());
        } catch (Exception e) {
            logger.info("准备消息发送失败 keys >>>>  " + keys, e);
        }
    }

    private String setTopicName(String topicName, String tags, String keys) {
        if (StringUtils.isNotBlank(TOP_NAME_PREFIX) && TOP_NAME_PREFIX.endsWith("_")) {
            topicName = TOP_NAME_PREFIX + topicName;
        } else {
            topicName = TOP_NAME_PREFIX + "_" + topicName;
        }
        logger.info("准备消息发送 topicName >>> " + topicName + ", tags >>>> " + tags + " , keys>>>>" + keys);
        return topicName;
    }
}


